package com.ctbri.manage.quality.example;
import com.ctbri.manage.quality.config.YamlReader;
import com.ctbri.manage.quality.multi.DifferenceStatic2;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

public class TwoTablesTest {
    public static void main( String[] args ) {
        //判断差集
        //导入yaml配置信息
        final String filePath = "/Users/songyunlong/Desktop/data-manage/data-quality/src/main/resources/config/sub/twoTablesTestConfig.yaml";
        Map<String, Object> conf = YamlReader.getInstance(filePath).getConf();
        System.out.println(conf);
        final String appName = YamlReader.getInstance(filePath).getString("appName");
        final String masterSet = YamlReader.getInstance(filePath).getString("masterSet");
        final String logLevel = YamlReader.getInstance(filePath).getString("logLevel");
        final String dataExportPath = YamlReader.getInstance(filePath).getString("dataExportPath");
        final String dataExportPath2 = YamlReader.getInstance(filePath).getString("dataExportPath2");
        final String dataSavePath = YamlReader.getInstance(filePath).getString("dataSavePath");
        final String dataSaveStyle = YamlReader.getInstance(filePath).getString("dataSaveStyle");
        final Boolean needStructType = YamlReader.getInstance(filePath).getBoolean("needStructType", false);
        final ArrayList<String> joinColName1 = YamlReader.getInstance(filePath).getList("joinColName1");
        final ArrayList<String> joinColName2 = YamlReader.getInstance(filePath).getList("joinColName2");
        final String joinMode = YamlReader.getInstance(filePath).getString("joinMode");
        final String dataSource = YamlReader.getInstance(filePath).getString("dataSource");
        final Boolean needPrefix = YamlReader.getInstance(filePath).getBoolean("needPrefix", false);

        DifferenceStatic2 differenceStatic2 = new DifferenceStatic2(
                appName,
                masterSet,
                logLevel,
                dataExportPath,
                dataSavePath,
                dataSaveStyle,
                dataExportPath2,
                joinColName1,
                joinColName2,
                joinMode
        );
        StructType structType1 = null;
        StructType structType2 = null;
        if (needStructType){
            final List<String> colName1 = YamlReader.getInstance(filePath).getList("colName1");
            List<StructField> structFields1 = new ArrayList<>();
            for (String i: colName1) structFields1.add(DataTypes.createStructField(i, DataTypes.StringType, true));
            structType1 = DataTypes.createStructType(structFields1);

            final List<String> colName2 = YamlReader.getInstance(filePath).getList("colName2");
            List<StructField> structFields2 = new ArrayList<>();
            for (String i: colName2) structFields2.add(DataTypes.createStructField(i, DataTypes.StringType, true));

            structType2 = DataTypes.createStructType(structFields2);
        }

        try{
            differenceStatic2.call(structType1, structType2, dataSource, true, true);
        } catch (Exception e){
            e.printStackTrace();
        }

    }
}
